package com.digiwin.demo.test;

import com.digiwin.demo.sink.AbstractElasticsearchSinkFunction;
import com.digiwin.demo.vo.VisitVO;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author yanggld
 * @Date 2019/12/26-14:24
 * @Description
 */
public class MyElasticsearchSinkFunction extends AbstractElasticsearchSinkFunction<VisitVO> {

	@Override
	public void invoke(VisitVO vo, RuntimeContext var2, RequestIndexer requestIndexer) {
		requestIndexer.add(createIndexRequest(vo));
	}

	public IndexRequest createIndexRequest(VisitVO vo) {
		Map<String, Object> json = new HashMap<>();
		json.put("uid", vo.getUid());
		json.put("name", vo.getName());
		json.put("count", vo.getCount());
		json.put("date", vo.getDate());
		return Requests.indexRequest()
				.index("book")
				.type("book_type")
				.source(json);
	}
}
